package cn.sciento.farm.devicemanager.gateway.config;



import cn.sciento.farm.devicemanager.gateway.uplink.handler.DeviceMessageHandler;
import cn.sciento.farm.devicemanager.gateway.uplink.handler.SysMessageHandler;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.ExecutorChannel;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.stereotype.Component;

import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;


@Component
@Slf4j
@Configuration
public class MqttConfig {

    @Autowired
    private MqttProperties mqttProperties;

    @Bean(name = "mqttCollectionFactory")
    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        MqttConnectOptions options = new MqttConnectOptions();
        options.setServerURIs(new String[]{mqttProperties.getServer()});
        options.setUserName(mqttProperties.getUsername());
        options.setPassword(mqttProperties.getPassword().toCharArray());
        factory.setConnectionOptions(options);
        return factory;
    }


    // 消费者
    @Bean
    @ServiceActivator(inputChannel = "mqttDeviceInputChannel")
    public MessageHandler deviceHandler() {
        return new DeviceMessageHandler();

    }

    @Bean(name = "mqttDeviceInputChannel")
    public MessageChannel mqttDeviceInputChannel() {
        return new ExecutorChannel(new ThreadPoolExecutor(5, 8,
                0L, TimeUnit.MILLISECONDS,
                //使用一个基于FIFO排序的阻塞队列，在所有corePoolSize线程都忙时新任务将在队列中等待
                new LinkedBlockingQueue<>()));
    }

    @Bean(name = "mqttDeviceInbound")
    public MessageProducerSupport mqttDeviceInbound() {
        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(
                mqttProperties.getServer(),
                mqttProperties.getSubclientId(),
                mqttClientFactory()
        );
        for (String topic : mqttProperties.getMainTopic()) {
            adapter.addTopic(topic);
        }
        adapter.setCompletionTimeout(5000);
        adapter.setOutputChannel(mqttDeviceInputChannel());
        DefaultPahoMessageConverter messageConverter = new DefaultPahoMessageConverter();
        messageConverter.setPayloadAsBytes(true);
        adapter.setConverter(messageConverter);
        adapter.setQos(mqttProperties.getQos());
        return adapter;
    }




    //sys
    @Bean
    @ServiceActivator(inputChannel = "mqttSysInputChannel")
    public MessageHandler SysHandler() {
        return new SysMessageHandler();

    }

    @Bean(name = "mqttSysInputChannel")
    public MessageChannel mqttSysInputChannel() {
        return new DirectChannel();
    }

    @Bean(name = "mqttSysInbound")
    public MessageProducerSupport mqttSysInbound() {
        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(
                mqttProperties.getServer(),
                mqttProperties.getMainclientId(),
                mqttClientFactory()
        );
        for (String topic : mqttProperties.getSysTopic()) {
            adapter.addTopic(topic);
        }
        adapter.setCompletionTimeout(mqttProperties.getTimeout());
        adapter.setOutputChannel(mqttSysInputChannel());
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(mqttProperties.getQos());
        return adapter;
    }



    // 生产者

    @Bean(name = "mqttsend")
    public MessageChannel outChannel() {
        return new DirectChannel();
    }

    @Bean
    @ServiceActivator(inputChannel = "mqttsend")
    public MqttPahoMessageHandler mqttOutbound() {
        MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(
                mqttProperties.getPubclientId(), mqttClientFactory());
//       messageHandler.setAsync(true);
        return messageHandler;
    }

}
